package com.example.java_message_queue.mqserver.core;

import com.example.java_message_queue.common.Consumer;
import com.example.java_message_queue.common.MqException;
import com.example.java_message_queue.mqserver.VirtualHost;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

// 实现消费功能
public class ConsumerManager {
    private VirtualHost parent;
    // 存放令牌的队列. 通过令牌来触发消费线程的消费操作.
    private BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);

    public ConsumerManager(VirtualHost parent) {
        this.parent = parent;

        // 启动扫描线程
        Thread scanThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String queueName = tokenQueue.take();
                    // 2. 找到队列
                    MSGQueue msgQueue = parent.getMemoryDataCenter().getQueue(queueName);
                    if (msgQueue == null) {
                        throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
                    }
                    // 3. 消费一个数据
                    synchronized (msgQueue) {
                        consumeMessage(msgQueue);
                    }
                } catch (MqException | InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "scanThread");
        scanThread.start();
    }

    private void consumeMessage(MSGQueue msgQueue) throws MqException {
        // 1. 按照轮询方式, 先找个消费者出来
        ConsumerEnv luckyDog = msgQueue.chooseConsumer();
        if (luckyDog == null) {
            // 如果当前还没有订阅者, 就先暂时不消费.
            return;
        }
        // 2. 从指定队列中取一个元素
        Message message = parent.getMemoryDataCenter().pollMessage(msgQueue.getName());
        if (message == null) {
            return;
        }
        System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + msgQueue.getName() + ", messageId=" + message.getMessageId());
        // 3. 丢到线程池中干活. 回调执行时间可能比较长. 不适合让扫描线程去调用.
        workerPool.submit(() -> {
            try {
                // 1. 先把消息放到待确认队列中
                //    (这个逻辑必须放到执行回调前面. 如果是 autoAck false, 在回调内部会调用 basicAck, 执行彻底删除. 需要先放到待确认队列, 才能彻底删除)
                parent.getMemoryDataCenter().addMessageWaitAck(msgQueue.getName(), message);
                // 2. 调用消费者的回调. 如果回调抛出异常了, 则不会对消息进行任何 ack 操作.
                //    相当于消息仍然处在待消费的状态.
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 3. 如果消息是自动确认, 则可以直接把消息彻底删除了.
                //    (这个逻辑必须放到执行回调后面. 万一执行回调一半服务器崩溃, 这个消息仍然存在于硬盘上, 下次启动还可以被继续消费到)
                if (luckyDog.isAutoAck()) {
                    //  则修改硬盘上的消息为 "无效". 同时删除内存中的消息
                    if (message.getDeliveryMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(msgQueue, message);
                    }
                    parent.getMemoryDataCenter().removeMessageWaitAck(msgQueue.getName(), message.getMessageId());
                    parent.getMemoryDataCenter().removeMessage(message.getMessageId());
                }
            } catch (MqException | IOException | ClassNotFoundException e) {
                e.printStackTrace();
            }
        });
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 消费已经积压的消息消息
        MSGQueue msgQueue = parent.getMemoryDataCenter().getQueue(queueName);
        if (msgQueue == null) {
            throw new MqException("[ConsumerManager] 队列不存在! queueName=" + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        synchronized (msgQueue) {
            msgQueue.addConsumerEnv(consumerEnv);

            // 把已经积压的 n 个数据都先消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                consumeMessage(msgQueue);
            }
        }
    }

    // 通知消费者去消费消息
    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }
}
